package com.dark.trident;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import storm.trident.operation.BaseFunction;
import storm.trident.operation.TridentCollector;
import storm.trident.tuple.TridentTuple;

import java.util.ArrayList;
import java.util.List;

/**
 * Created by tengxue on 16-7-1.
 */
public class OutbreakDetector extends BaseFunction {

    private static final long serialVersionUID=1L;

    private static final Logger LOG= LoggerFactory.getLogger(OutbreakDetector.class);

    private static final int THRESHOLD=10000;



    public void execute(TridentTuple tuple, TridentCollector collector) {
        String key=(String)tuple.getValue(0);
        Long count=(Long) tuple.getValue(1);
        LOG.info("the key is:"+key+" and count is ==> [ "+count+" ]");
        if (count>THRESHOLD){
            List<Object> values=new ArrayList<Object>();
            values.add("Outbreak detected for ["+ key+" ]");
            collector.emit(values);
        }
    }
}
